package ed2.fisico.juncao;

import java.util.Hashtable;

import ed2.base.Operacao;
import ed2.fisico.acesso.ScanTabela;
import ed2.fisico.acesso.Tupla;

public class HashJoin implements Operacao {

	private Operacao relacaoConstrucao;
	private Operacao relacaoPoda;
	private String campoRelacao2;
	private String campoRelacao1;
	private Hashtable<String, Tupla> tabela;
	
	public HashJoin(String relacao1, String campoRelacao1, String relacao2, String campoRelacao2) {
		this.relacaoConstrucao = new ScanTabela(relacao1);
		this.relacaoPoda = new ScanTabela(relacao2);
		this.campoRelacao1 = campoRelacao1;
		this.campoRelacao2 = campoRelacao2;
	}
	
	public HashJoin(Operacao relacao1, String campoRelacao1, Operacao relacao2, String campoRelacao2) {
		this.relacaoConstrucao = relacao1;
		this.relacaoPoda = relacao2;
		this.campoRelacao1 = campoRelacao1;
		this.campoRelacao2 = campoRelacao2;
	}

	
	@Override
	public Operacao abrir() {
		tabela = new Hashtable<>();
		relacaoConstrucao.abrir();
		Tupla tupla;
		while ((tupla = (Tupla) relacaoConstrucao.ler()) != null) {
			String valor = tupla.getValorCampo(campoRelacao1);
			tabela.put(valor, tupla);
		}
		relacaoPoda.abrir();
		return this;
	}

	@Override
	public Operacao ler() {
		Tupla tupla = (Tupla) relacaoPoda.ler();
		if (tupla == null) {
			return null;
		}
		Tupla tuplaCorrespondente = tabela.get(tupla.getValorCampo(campoRelacao2));
		if(tuplaCorrespondente == null){
			return new Tupla();
		}
		else{
			Tupla tuplaResultante = new Tupla();
			tuplaResultante.concatena(tupla);
			tuplaResultante.concatena(tuplaCorrespondente);
			return tuplaResultante;
		}
	}

	@Override
	public Operacao fechar() {
		relacaoConstrucao.fechar();
		relacaoPoda.fechar();
		return this;
	}
	
	@Override
	public long calcularCusto() {
		return relacaoConstrucao.calcularCusto() + relacaoPoda.calcularCusto();
	}

}
